package com.xiaochong.bitong.job.config.hbase;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.xc.logclient.utils.LogTrace;
import com.xiaochong.service.common.config.redis.RedisServer;
import com.xiaochong.tir.common.exception.hbase.HbaseCloseExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseConnectionExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseGetTableExcetion;
import com.xiaochong.tir.common.exception.hbase.HbaseTablePutExcetion;
import com.xiaochong.tir.common.util.HbaseTableUtils;
import org.apache.hadoop.hbase.client.Connection;
import org.apache.hadoop.hbase.client.Put;
import org.apache.hadoop.hbase.client.Table;
import org.apache.hadoop.hbase.util.Bytes;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.util.*;


/**
 * Created by cx on 2018/4/25.
 */
@Component
public class CurrencyInfoHbaseComponent {

    @Autowired
    private HbaseConnectionFactory hbaseConnectionFactory;

    @Autowired
    private HbaseBaseQueryComponent hbaseBaseQueryComponent;

    @Autowired
    private RedisServer redisServer;

    /**
     * 示例
     */
    public void test(){
        try {

            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, HbaseTableUtils.CURRENCY_INFO_TABLE_NAME);
            Put put = new Put("20180428".getBytes());
            put.addColumn(HbaseTableUtils.CURRENCY_INFO_TABLE_FAMILY.getBytes(),"qualifier".getBytes(),new JSONObject().toJSONString().getBytes());
            hbaseConnectionFactory.tablePut(table,put);
            hbaseConnectionFactory.closeAll(connection,table);


        } catch (HbaseConnectionExcetion hbaseConnectionExcetion) {
            hbaseConnectionExcetion.printStackTrace();
        }  catch (HbaseGetTableExcetion hbaseGetTableExcetion) {
            hbaseGetTableExcetion.printStackTrace();
        } catch (HbaseTablePutExcetion hbaseTablePutExcetion) {
            hbaseTablePutExcetion.printStackTrace();
        } catch (HbaseCloseExcetion hbaseCloseExcetion) {
            hbaseCloseExcetion.printStackTrace();
        }
    }



    /**
     * 插入hbase数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param qualifier
     * @param value
     */
    public void insertHbaseString(String tableName, String rowKey, String family, String key, String value){
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            Put put = new Put(rowKey.getBytes());
            put.addColumn(Bytes.toBytes(family),
                    Bytes.toBytes(key),
                    Bytes.toBytes(value));
            hbaseConnectionFactory.tablePut(table,put);
            hbaseConnectionFactory.closeAll(connection,table);

        } catch (HbaseConnectionExcetion | HbaseGetTableExcetion | HbaseCloseExcetion | HbaseTablePutExcetion e) {
            LogTrace.error(e.getMessage(), e);
        }
    }


    /**
     * 插入hbase数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param object
     */
    public void insertHbaseObject(String tableName, String rowKey, String family,Object object) {
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            Put put = new Put(rowKey.getBytes());
            JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(object));
            Map<String, Object> map = jsonObject.getInnerMap();
            for(Map.Entry<String, Object> entry: map.entrySet()){
                String key = entry.getKey();
                put.addColumn(Bytes.toBytes(family),
                        Bytes.toBytes(key),
                        Bytes.toBytes(entry.getValue()!=null?entry.getValue().toString():"") );
            }
            hbaseConnectionFactory.tablePut(table,put);
            hbaseConnectionFactory.closeAll(connection,table);
        } catch (HbaseCloseExcetion | HbaseGetTableExcetion | HbaseTablePutExcetion | HbaseConnectionExcetion e) {
            e.printStackTrace();
            LogTrace.error(e.getMessage(),e);
        }
    }



    /**
     * hbase批量插入数据
     *
     * @param tableName
     * @param rowKey
     * @param family
     * @param list
     */
    public void insertHbaseList(String tableName, String rowKey, String family,List list) {
        try {
            Connection connection = hbaseConnectionFactory.getConnection();
            Table table =  hbaseConnectionFactory.getTable( connection, tableName);
            List<Put> putList = new ArrayList<>();
            for (int i = 0; i < list.size(); i++) {
                Put put = new Put(rowKey.getBytes());//row kye
                JSONObject jsonObject = JSON.parseObject(JSON.toJSONString(list.get(i)));
                Map<String, Object> map = jsonObject.getInnerMap();
                for(Map.Entry<String, Object> entry: map.entrySet()){
                    String key = entry.getKey();
                    put.addColumn(Bytes.toBytes(family),
                            Bytes.toBytes(key),
                            Bytes.toBytes(entry.getValue()!=null?entry.getValue().toString():"") );
                }

                putList.add(put);
            }
            hbaseConnectionFactory.tablePut(table,putList);
            hbaseConnectionFactory.closeAll(connection,table);
        } catch (HbaseCloseExcetion | HbaseGetTableExcetion | HbaseTablePutExcetion | HbaseConnectionExcetion e) {
            e.printStackTrace();
            LogTrace.error(e.getMessage(),e);
        }
    }



}
